﻿using System;
using System.Collections.Generic;

using Apache.Cassandra;
using Client = Apache.Cassandra.Cassandra.Client;

using RexToy.Cassandra.Meta;

namespace RexToy.Cassandra
{
    class ColumnFamily<K, C> : ColumnFamilyBase<K, C>, IColumnFamily<K, C>
    {
        public ColumnFamily(string keyspace, string columnFamily, ColumnFamilyInfo info, Client c, ConsistencyLevel read, ConsistencyLevel write)
            : base(keyspace, columnFamily, info, c, read, write)
        {
        }

        public void Insert(K key, C column, object value)
        {
            var cp = Utils.CreateColumnParent(this.Name);
            var c = new Apache.Cassandra.Column();
            c.Name = this.GetColumnBytes(column);
            c.Value = this.GetValueBytes(column, value);
            c.Timestamp = this.GetTimeStamp();

            _c.insert(this.GetKeyBytes(key), cp, c, _write);
        }

        public void Remove(K key, C column)
        {
            ColumnPath cp = Utils.CreateColumnPath(this.Name, this.GetColumnBytes(column));
            _c.remove(this.GetKeyBytes(key), cp, this.GetTimeStamp(), _write);
        }

        public void Remove(K key)
        {
            ColumnPath cp = Utils.CreateColumnPath(this.Name, null);
            _c.remove(this.GetKeyBytes(key), cp, this.GetTimeStamp(), _write);
        }

        public object Select(K key, C column)
        {
            ColumnPath cp = Utils.CreateColumnPath(this.Name, this.GetColumnBytes(column));
            var c = _c.get(this.GetKeyBytes(key), cp, _read);
            return this.GetValueData(column, c.Column.Value);
        }

        public IRow<K, C> Select(K key, int count = 10, bool reversed = false)
        {
            return this._Select(key, default(C), default(C), count, reversed);
        }

        public IRow<K, C> Select(K key, C start, C finish, int count = 10, bool reversed = false)
        {
            return this._Select(key, start, finish, count, reversed);
        }

        private IRow<K, C> _Select(K key, C start, C finish, int count, bool reversed)
        {
            SlicePredicate predicate = Utils.CreateSlicePredicate(this.GetColumnBytes(start), this.GetColumnBytes(finish), count, reversed);
            ColumnParent cp = Utils.CreateColumnParent(this.Name);

            var list = _c.get_slice(this.GetKeyBytes(key), cp, predicate, _read);

            var collection = this.CreateColumns(list);
            return this.CreateRow(key, collection);
        }

        public IEnumerable<IRow<K, C>> Select(IEnumerable<K> keys, int count = 10, bool reversed = false)
        {
            return this._Select(keys, default(C), default(C), count, reversed);
        }

        public IEnumerable<IRow<K, C>> Select(IEnumerable<K> keys, C start, C finish, int count = 10, bool reversed = false)
        {
            return this._Select(keys, start, finish, count, reversed);
        }

        private IEnumerable<IRow<K, C>> _Select(IEnumerable<K> keys, C start, C finish, int count, bool reversed)
        {
            SlicePredicate predicate = Utils.CreateSlicePredicate(this.GetColumnBytes(start), this.GetColumnBytes(finish), count, reversed);
            ColumnParent parent = Utils.CreateColumnParent(this.Name);

            List<byte[]> byteKeys = this.ConvertKeyList(keys);
            var result = _c.multiget_slice(byteKeys, parent, predicate, _read);

            List<IRow<K, C>> rows = new List<IRow<K, C>>();
            foreach (var kvp in result)
            {
                var collection = this.CreateColumns(kvp.Value);
                var row = this.CreateRow(this.GetKeyData(kvp.Key), collection);
                rows.Add(row);
            }
            return rows;
        }

        public IEnumerable<IRow<K, C>> List(int rowCount = 100, int columnCount = 10, bool reversedColumn = false)
        {
            return _List(default(K), default(K), default(C), default(C), rowCount, columnCount, reversedColumn);
        }

        public IEnumerable<IRow<K, C>> List(C startColumn, C finishColumn, int rowCount = 100, int columnCount = 10, bool reversedColumn = false)
        {
            return _List(default(K), default(K), startColumn, finishColumn, rowCount, columnCount, reversedColumn);
        }

        public IEnumerable<IRow<K, C>> List(K startKey, K finishKey, C startColumn, C finishColumn, int rowCount = 100, int columnCount = 10, bool reversedColumn = false)
        {
            return _List(startKey, finishKey, startColumn, finishColumn, rowCount, columnCount, reversedColumn);
        }

        private IEnumerable<IRow<K, C>> _List(K startKey, K finishKey, C startColumn, C finishColumn, int rowCount, int columnCount, bool reversedColumn)
        {
            KeyRange range = Utils.CreateKeyRange(this.GetKeyBytes(startKey), this.GetKeyBytes(finishKey), rowCount);
            ColumnParent parent = Utils.CreateColumnParent(this.Name);
            SlicePredicate predicate = Utils.CreateSlicePredicate(this.GetColumnBytes(startColumn), this.GetColumnBytes(finishColumn), columnCount, reversedColumn);

            var result = _c.get_range_slices(parent, predicate, range, _read);

            var rows = new List<IRow<K, C>>();
            foreach (var r in result)
            {
                var coll = this.CreateColumns(r.Columns);
                var row = this.CreateRow(this.GetKeyData(r.Key), coll);
                rows.Add(row);
            }
            return rows;
        }

        #region Abstract method implement
        protected override IRow<K, C> CreateRow(K key, IColumnCollection<C> collection)
        {
            return new Row<K, C>(key, collection);
        }
        #endregion

        private IColumn<C> CreateColumn(Apache.Cassandra.Column column)
        {
            var col = this.GetColumnData(column.Name);
            return new Column<C>(col, this.GetValueData(col, column.Value));
        }

        private IColumnCollection<C> CreateColumns(List<ColumnOrSuperColumn> list)
        {
            var collection = new ColumnCollection<C>();
            foreach (var col in list)
            {
                collection.AddColumn(this.CreateColumn(col.Column));
            }
            return collection;
        }
    }
}
